package org.example.config.mqtt;

import com.google.common.collect.Lists;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.kafka.clients.admin.NewTopic;
import org.example.config.mqtt.property.MqttProperty;
import org.example.config.mqtt.property.MqttTopicProperty;
import org.example.enums.MqttChannelTypeEnum;
import org.example.enums.service.EnumsBaseService;
import org.example.utils.ManualRegistBeanUtil;
import org.example.utils.ValidateUtils;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.transformer.GenericTransformer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.List;
import java.util.Objects;

@Component
@Slf4j
@EnableConfigurationProperties(MqttProperty.class)
@RequiredArgsConstructor
@ConditionalOnBean(MqttConfig.class)
public class MqttTopicHandle {
    private final MqttProperty mqttProperty;

    private final MqttPahoClientFactory mqttClientFactory;
    private final MessageConverter messageConverter;
    private final ApplicationContext applicationContext;
    @Resource(name = "mqttInputErrorChannel")
    private MessageChannel errorChannel;

    @PostConstruct
    public void initTopic() {
        if (CollectionUtils.isEmpty(mqttProperty.getMqttTopics())) {
            log.info("mqtt topic is empty");
            return;
        }

        //校验参数
        ValidateUtils.validateObj(mqttProperty.getMqttTopics());
        //初始化topic
        mqttProperty.getMqttTopics().forEach(mqttTopicProperty -> {
            MqttPahoMessageDrivenChannelAdapter mqttPahoMessageDrivenChannelAdapter = mqttInboundAdapter(mqttClientFactory, errorChannel, mqttTopicProperty);
            ManualRegistBeanUtil.registerBean((ConfigurableApplicationContext) applicationContext, mqttPahoMessageDrivenChannelAdapter, mqttTopicProperty.getChannelAdapterName());
            IntegrationFlow integrationFlow = mqttInboundFlow(mqttPahoMessageDrivenChannelAdapter, messageConverter);
            ManualRegistBeanUtil.registerBean((ConfigurableApplicationContext) applicationContext,integrationFlow, mqttTopicProperty.getIntegrationFlowName());
        });
    }

    public MqttPahoMessageDrivenChannelAdapter mqttInboundAdapter(MqttPahoClientFactory mqttClientFactory, MessageChannel errorChannel, MqttTopicProperty mqttTopicProperty) {
        MessageChannel outPutChannel = mqttTopicProperty.getOutPutChannelType().build(mqttTopicProperty.getOutPutChannelName());
        ManualRegistBeanUtil.registerBean((ConfigurableApplicationContext) applicationContext, outPutChannel, mqttTopicProperty.getOutPutChannelName());
        List<Integer> qosList = Lists.newArrayListWithCapacity(mqttTopicProperty.getTopics().size());
        List<String> topics = Lists.newArrayListWithCapacity(mqttTopicProperty.getTopics().size());
        mqttTopicProperty.getTopics()
                .forEach(topic -> {
                    qosList.add(Objects.isNull(topic.getQos()) ? 0 : topic.getQos());
                    topics.add(topic.getTopic());
                });
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                mqttTopicProperty.getClientId(),
                mqttClientFactory,
                topics.toArray(new String[0]));
        adapter.setCompletionTimeout(Objects.isNull(mqttTopicProperty.getCompletionTimeout()) ? 5000 : mqttTopicProperty.getCompletionTimeout());
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(qosList.stream().mapToInt(Integer::intValue).toArray());
        TaskScheduler taskScheduler = build(mqttTopicProperty.getChannelAdapterName());
        adapter.setTaskScheduler(taskScheduler);
        //手动ack
        adapter.setManualAcks(mqttTopicProperty.isManualAcks());
        adapter.setOutputChannel(outPutChannel);
        //错误处理，重试机制用完还是失败的消息，连接异常
        adapter.setErrorChannel(errorChannel);
        return adapter;
    }

    private TaskScheduler build(String channelAdapterName){
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setPoolSize(2); // 设置线程池大小
        scheduler.setThreadNamePrefix("mqtt-"+channelAdapterName+"-taskScheduler-"); // 设置线程名前缀
        scheduler.initialize();
        return scheduler;
    }
    public IntegrationFlow mqttInboundFlow(MqttPahoMessageDrivenChannelAdapter mqttPahoMessageDrivenChannelAdapter, MessageConverter messageConverter) {
        return IntegrationFlows
                .from(mqttPahoMessageDrivenChannelAdapter)
                .transform((GenericTransformer<Message<?>, ?>) source -> {
                    Object clazz = source.getHeaders().get("clazz");
                    if (Objects.isNull(clazz)) {
                        return source.getPayload();
                    }
                    return messageConverter.fromMessage(source, (Class<?>) clazz);
                })
                .get();
    }
}
